package com.salmon.rpc.utils;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoPool;
import de.javakaffee.kryoserializers.SynchronizedCollectionsSerializer;
import org.objenesis.strategy.StdInstantiatorStrategy;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * kryo的对象本身不是线程安全的，所以我们有两种选择来保障线程安全。
 * 使用Threadlocal来保障线程安全:
 * private static final ThreadLocal<Kryo> kryoLocal = new ThreadLocal<Kryo>() {
 *     protected Kryo initialValue() {
 *         Kryo kryo = new Kryo();
 *         kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(
 *                     new StdInstantiatorStrategy()));
 *         return kryo;
 *     };
 * };
 *
 * 或者使用kryo提供的pool:
 * public KryoPool newKryoPool() {
 *         return new KryoPool.Builder(() -> {
 *             final Kryo kryo = new Kryo();
 *             kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(
 *                     new StdInstantiatorStrategy()));
 *             return kryo;
 *         }).softReferences().build();
 *     }
 *
 */
public class KryoUtils {

    public static KryoPool newKryoPool() {
        return new KryoPool.Builder(() -> {
            final Kryo kryo = new Kryo();
            kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(
                    new StdInstantiatorStrategy()));
            //支持对象循环引用（否则会栈溢出）
            //kryo.setReferences(true); //默认值就是 true，添加此行的目的是为了提醒维护者，不要改变这个配置
            //不强制要求注册类（注册行为无法保证多个 JVM 内同一个类的注册编号相同；而且业务系统中大量的 Class 也难以一一注册）
            //kryo.setRegistrationRequired(false); //默认值就是 false，添加此行的目的是为了提醒维护者，不要改变这个配置
            //在系列化有Hashtable、Hashmap之类的对象属性的时候，会遇到异常 maven中引入 kryo-serializers 并执行下面代码
            SynchronizedCollectionsSerializer.registerSerializers(kryo);
            return kryo;
        }).softReferences().build();
    }

    public static <T> byte[] serialize(T obj) {
        try {
            KryoPool pool = newKryoPool();
            Kryo kryo = pool.borrow();
            try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
                Output output = new Output(os);
                kryo.writeObject(output, obj);
                output.flush();
                pool.release(kryo);// 释放
                return os.toByteArray();
            }
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public static <T> T deserialize(byte[] bytes, Class<T> clazz) {
        KryoPool pool = newKryoPool();
        Kryo kryo = pool.borrow();
        try(ByteArrayInputStream is = new ByteArrayInputStream(bytes)) {
            Input input = new Input(is);
            T t = kryo.readObject(input, clazz);
            pool.release(kryo);// 释放
            return t;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }
}
